package org.apache.spark.rpc.netty

import java.util.concurrent.TimeUnit

import org.apache.spark.rpc.{RpcAddress, RpcEndpointRef, RpcEnv}
import org.apache.spark.{SecurityManager, SparkConf}

class RpcClient(remoteEndpointRef: RpcEndpointRef) {
  // 发送双向消息
  def testAdd(a: Int, b: Int): Int = {
    remoteEndpointRef.askSync[Result](TestAdd(a, b)).result
  }
  def testSub(a: Int, b: Int): Int = {
    remoteEndpointRef.askSync[Result](TestSub(a, b)).result
  }
  // 发送单向消息
  def testSend() = {
    remoteEndpointRef.send(HelloMessage("hello"))
  }
}
object RpcClient {

  def main(args: Array[String]): Unit = {
    println("TestRpcEnv")
    val conf = new SparkConf()
    val isDriver = false
    val systemName = "sparkDriver"
    val hostname = "127.0.0.1"
    val port = 4040
    val isLocal = true

    val securityManager = new SecurityManager(conf)
    // 创建rpc context
    val rpcEnv: NettyRpcEnv = RpcEnv.create(systemName, hostname, port, conf, securityManager, clientMode = true).asInstanceOf[NettyRpcEnv]

    // 创建远程服务的引用，即客户端
    val remoteEndpointRef = rpcEnv.setupEndpointRef(RpcAddress(hostname, 4040), RpcServer.SERVER_ENDPOINT_NAME)

    val testMathMaster = new RpcClient(remoteEndpointRef)
    val result = testMathMaster.testAdd(1, 2)
    println("the result of test: " + result)
    testMathMaster.testSend()

    TimeUnit.SECONDS.sleep(10000)
  }

}
